package cn.lhz.italy.etl

import cn.lhz.util.spark.SparkUtil
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.{col, regexp_replace, when}
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType}

/**
 * 对 italy.csv 进行数据清洗，过滤掉包含空值的行，并保存至 Hive 表
 *
 * @author 胡书源
 * @version 1.0.0
 * 改（年阔正）
 * @version 1.0.1
 *
 */
object ItalyEtl {
  def main(args: Array[String]): Unit = {
    // 获取 SparkSession 对象
    val sparkSession = SparkUtil()

    // 定义源数据在HDFS文件系统的路径
    val originPath: String = "/air-data/ods/AirItaly.csv"

    // 读取 CSV 文件，获取 DataFrame
    val df = sparkSession.read.option("header", "true").csv(originPath)

    // 字段集合（CSV 文件的列名）
    val cols = Array(
      "date",
      "time",
      "co_gt",            // 将 "co(gt)" 改为 "co_gt"
      "pt08_s1_co",       // 将 "pt08_s1(co)" 改为 "pt08_s1_co"
      "nmhc_gt",          // 将 "nmhc(gt)" 改为 "nmhc_gt"
      "c6h6_gt",          // 将 "c6h6(gt)" 改为 "c6h6_gt"
      "pt08_s2_nmhc",     // 将 "pt08_s2(nmhc)" 改为 "pt08_s2_nmhc"
      "nox_gt",           // 将 "nox(gt)" 改为 "nox_gt"
      "pt08_s3_nox",      // 将 "pt08_s3(nox)" 改为 "pt08_s3_nox"
      "no2_gt",           // 将 "no2(gt)" 改为 "no2_gt"
      "pt08_s4_no2",      // 将 "pt08_s4(no2)" 改为 "pt08_s4_no2"
      "pt08_s5_o3",       // 将 "pt08_s5(o3)" 改为 "pt08_s5_o3"
      "t",
      "rh",
      "ah"
    )

    // 过滤掉有空值的数据行
    val cleanedDf = df.na.drop(cols)




    // 为字段指定合适的数据类型（根据实际情况进行调整）
    val table = cleanedDf.select(
      col("date"),
      col("time").cast(StringType),           // 将 "time" 字段的数据类型改为 StringType
      col("co_gt").cast(DoubleType),           // 替换 "co(gt)" 为 "co_gt"
      col("pt08_s1_co").cast(IntegerType),     // 替换 "pt08_s1(co)" 为 "pt08_s1_co"
      col("nmhc_gt").cast(IntegerType),         // 替换 "nmhc(gt)" 为 "nmhc_gt"
      col("c6h6_gt").cast(DoubleType),         // 替换 "c6h6(gt)" 为 "c6h6_gt"
      col("pt08_s2_nmhc").cast(IntegerType),   // 替换 "pt08_s2(nmhc)" 为 "pt08_s2_nmhc"
      col("nox_gt").cast(IntegerType),          // 替换 "nox(gt)" 为 "nox_gt"
      col("pt08_s3_nox").cast(IntegerType),    // 替换 "pt08_s3(nox)" 为 "pt08_s3_nox"
      col("no2_gt").cast(IntegerType),          // 替换 "no2(gt)" 为 "no2_gt"
      col("pt08_s4_no2").cast(IntegerType),    // 替换 "pt08_s4(no2)" 为 "pt08_s4_no2"
      col("pt08_s5_o3").cast(IntegerType),     // 替换 "pt08_s5(o3)" 为 "pt08_s5_o3"
      col("t").cast(DoubleType),
      col("rh").cast(DoubleType),
      col("ah").cast(DoubleType)
    )


    // 保存数据到 Hive 表
    table.write.mode(SaveMode.Overwrite).saveAsTable("air.dwd_italy")

    // 释放资源
    sparkSession.stop()

  }
}